package com.atguigu.orderpay_detect

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @Author: ys
 * @Description:
 * @Date: create in 2020/5/15 14:18
 */

case class ReceiptEvent(txId: String, payChannel: String, eventTime: Long)

object TxMacthDetect {

  //定义侧输出流tag，输出异常数据
  //在订单支付流中找到了，没有在对账流中找到
  val unMatchedPays = new OutputTag[OrderEvent]("unMatchedPays")
  //只拿到对账流事件，但是没有匹配的orderpay支付事件
  val unMatchedReceipt = new OutputTag[ReceiptEvent]("unMatchedReceipt")

  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取订单事件流
    val resource = getClass.getResource("/OrderLog.csv")
    val orderEventStream = environment.readTextFile(resource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
      })
      .filter(_.txId != "")
      //设置时间戳
      //方式一(简单实现)：升序数据，不设置延时 到点就发车，从字段中抽取
      .assignAscendingTimestamps(_.eventTime * 1000L)
      .keyBy(_.txId)

    //读取支付到账事件流
    val receiptResource = getClass.getResource("/ReceiptLog.csv")
    val receiptEventStream = environment.readTextFile(receiptResource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        ReceiptEvent(dataArray(0).trim, dataArray(1).trim, dataArray(2).trim.toLong)
      })
      .assignAscendingTimestamps(_.eventTime * 1000L)
      //和前面orderEventStream处理保持一致，
      .keyBy(_.txId)

    //将两条流连接起来，共同处理
    val processStream = orderEventStream.connect(receiptEventStream)
        .process(new TxPayMacth())

    processStream.print("matched")
    processStream.getSideOutput(unMatchedPays).print("unMatchedPays")
    processStream.getSideOutput(unMatchedReceipt).print("unMatchedReceipt")

    environment.execute()

  }

  class TxPayMacth() extends CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]{

    //定义状态用来保存已经到达的订单支付事件
    lazy val payState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor[OrderEvent]("pay-state",classOf[OrderEvent]))
    //定义状态用来保存到账事件
    lazy val receiptState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receipt-state",classOf[ReceiptEvent]))

    //第一条流orderEventStream的操作，订单支付事件数据的处理
    override def processElement1(pay: OrderEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      //判断有没有对应的到账事件
      val receipt = receiptState.value()
      if (null != receipt){
        //如果已经有receipt，在主输出流输出匹配信息，清空状态
        out.collect((pay, receipt))
        //payState没有保存过，所以不用清空
        receiptState.clear()
      }else {
        //如果到账事件还没到，那么把pay存入状态，并注册一个定时器等待
        payState.update(pay)
        ctx.timerService().registerEventTimeTimer(pay.eventTime * 1000L + 5000L)  //延时5秒，等待到账事件
      }
    }
    //第二条流receiptEventStream，到账事件的处理
    override def processElement2(receipt: ReceiptEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      //同样处理流程
      val pay = payState.value()
      if (null != pay){
        out.collect((pay, receipt))
        payState.clear()
      }else {
        receiptState.update(receipt)
        ctx.timerService().registerEventTimeTimer(receipt.eventTime * 1000L + 5000L)
      }
    }

    override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      //到时间了，如果还没有收到某个事件，那么输出报警信息

      if (payState.value() != null){
        //receipt没来，没有到账事件，输出pay到侧输出流
        ctx.output(unMatchedPays, payState.value())
      }
      if (receiptState.value() != null){
        ctx.output(unMatchedReceipt, receiptState.value())
      }
      payState.clear()
      receiptState.clear()
    }
}

}
